/*
 * Heartbeat.java
 *
 * Created on May 2, 2002, 1:56 PM
 */

package redman;

import java.net.*;
import java.util.*;
import java.text.*;
import java.lang.*;
import java.io.*;
import org.apache.log4j.*;

/**
 *
 * @author  llaranje
 * @version 
 */

public class Heartbeat {
    
    String name;
    String localhost;
    String recvhost;
    String sendhost;
    boolean running = false;
    boolean backup = false;
    boolean go = true;
    boolean torun = false;
    boolean initok = true;
    HbRecv hbrecv;
    HbSend hbsend;
    int portsend = 0;
    int portrecv = 0;
    int hbint = 0;
    int secfactor = 0;
    Object sem = null;
    String event;
    
    TimerTask sendTask = null;
    
    private static Category trace = Category.getInstance(Heartbeat.class);
    
    public void cleanup() {
        hbrecv.s.close();
        hbsend.s.close();
        go = false;
    }
    
    //
    // This method verifies if the local host is in the  host  list  defined
    // for this heartbeat. If it is, it returns true and also  collects  the
    // name of the remote host to which it will  send  heartbeats,  and  the
    // remote host from which it will receive heartbeats.
    //
    // The host to which the local host will send  heartbeats  (sendhost) is 
    // the next in the list.
    //
    // The host from which the local host will receive heartbeats (recvhost)
    // is the previous the list.
    //
    // If the local host is not in the host list for  this  heartbeat,  this
    // method returns false.
    //
    boolean hbToRunOnThisHost() throws UnknownHostException {
       // Declarations of auxiliary variables
       String hostname = null;
       String firsthost = null;       
       String lasthost = null;
       String previoushost = null;
       boolean hostinlist = false;
       
       localhost = InetAddress.getLocalHost().getHostName(); 
       String hosts = RedMan.cfgtable.getProperty(name + "_HOSTS");
       StringTokenizer st = new StringTokenizer(hosts);
       
       if (st.hasMoreTokens() == false) {
           trace.error(name + ": Bad config file. No host names in heartbeat definition.");
           return(false);
       }
       
       while (st.hasMoreTokens()) {
           hostname = st.nextToken();
           if (firsthost == null) firsthost = hostname;
           if (localhost.equals(hostname)) {
               hostinlist = true;
               if (previoushost != null) recvhost = previoushost;
               if (st.hasMoreTokens() == true) {
                   hostname = st.nextToken();
                   sendhost = hostname;
               }
               if ((recvhost != null) && (sendhost != null)) break;
           }
           if (st.hasMoreTokens() == false) lasthost = hostname;
           else previoushost = hostname; 
       }
       
       if (firsthost.equals(lasthost)) {
           trace.error(name + ": Bad config file. Only one host defined for heartbeat");
       }
       else {
           if (hostinlist) {
              if (recvhost == null) recvhost = lasthost;
              if (sendhost == null) sendhost = firsthost;
           }
       }
       
       return(hostinlist);
    }
    
    public Heartbeat (String hbname) throws UnknownHostException {

        // Save the name of the heartbeat object
        name = hbname;
        
        // Get localhost name
        localhost = InetAddress.getLocalHost().getHostName(); 
    
        trace.info("Heartbeat " + hbname + " being instantiated on host " + localhost);

        // Create semaphore object
        sem = new Object(); 
        trace.debug(name + ": semaphore instantiated");
                
        // Only start threads if the local host is in the host list
        // defined for this heartbeat
        if (hbToRunOnThisHost()) {
            torun = true;
            // Print message
            trace.debug(name + ": on host " + localhost + " receives heartbeats from host " 
                                            + recvhost + " and sends heartbeats to host " + sendhost);
            
            // Get characterisitcs of the heartbeat
            String stperiod = RedMan.cfgtable.getProperty(name + "_PERIOD");
            if (stperiod == null) {
                trace.error(name + ": Invalid configuration file. Heartbeat period not specified");
                initok = false;
                return;
            }
            else stperiod = stperiod.trim();
            hbint = Integer.parseInt(stperiod);
            String stport = RedMan.cfgtable.getProperty(name + "_PORT");
            if (stport == null) {
                trace.error(name + ": Invalid configuration file. Heartbeat port not specified");
                initok = false;
                return;
            }
            else stport = stport.trim();
            portsend = Integer.parseInt(stport);
            portrecv = portsend;
            trace.debug(name + " (primary) ports: recv = " + String.valueOf(portrecv)
                                                           + ", send = " + String.valueOf(portsend));
            String stsecfactor = RedMan.cfgtable.getProperty(name + "_SECURITY_FACTOR");
            if (stsecfactor == null) {
                trace.error(name + ": Invalid configuration file. Heartbeat security factor not specified");
                initok = false;
                return;
            }
            else stsecfactor = stsecfactor.trim();
            secfactor = Integer.parseInt(stsecfactor);
            
            // Create the receiving thread
            hbrecv = new HbRecv();
            trace.info(name + ": heartbeat obj created thread obj hbrecv");
            hbrecv.setPriority(Thread.MAX_PRIORITY);
        
            // Create the sending thread
            hbsend = new HbSend(); 
            trace.info(name + ": heartbeat obj created thread obj hbsend");
            hbsend.setPriority(Thread.MAX_PRIORITY);
        }
        else {
            trace.info(name + ": not running on host " + localhost);
        }
    }
    
    class HbRecv extends java.lang.Thread {
        boolean firsthb = true;
        DatagramSocket s;
        DatagramPacket p;
    
        /** Creates new HbRecv */
        public HbRecv() {}           

        public void run() {
            trace.debug(name + ": hbrecv running - hbint = " + String.valueOf(hbint));

            DateFormat time = null;
            long t1 = 0, t2 = 0;
        
            try {
                String msg;

                byte[] buffer = new byte[4096];
                s = new DatagramSocket(portrecv);
                s.setSoTimeout(0);
                p = new DatagramPacket(buffer, buffer.length);
                
                trace.debug(name + ": receive socket: bound to local port " 
                                                    + Integer.toString(s.getLocalPort())
                                                    + ", connected to port " + Integer.toString(s.getPort()));
            
                int i=1;
                do {
                    try {
                        time = DateFormat.getTimeInstance(DateFormat.FULL);
                        // trace.debug(name + ": hbrecv will wait to receive a heartbeat at " 
                        //                                             + time.format(new Date()));
                        s.receive(p);
                        time = DateFormat.getTimeInstance(DateFormat.FULL);
                        if (!name.equals("HBSELF")) {
                            trace.debug(name + ": hbrecv received a heartbeat at " + time.format(new Date()));
                        }
                    }
                    catch (java.io.IOException e) {
                        if (go == false) break;
                        if (e.getMessage().equals("Receive timed out")) {
                            t2 = System.currentTimeMillis();
                            trace.debug(name + ": hbrecv: Missed heartbeat after "
                                                    + Long.toString(t2 - t1) + " millisecs");
                        }
                        else {
                            trace.error(name + ": receive exception: " 
                                                        + e.getMessage());
                        }
                        s.setSoTimeout(0);
                        firsthb = true;
                    
                        event = recvhost + " DOWN";                       
                        trace.debug(name + ": hbrecv@" + localhost 
                                         + " will synchronize on semaphore - hostevent DOWN occurred");
                        
                        synchronized (sem) {
                            trace.debug(name + ": hbrecv@" + localhost + " will notify semaphore");
                            sem.notifyAll();
                            trace.debug(name + ": hbrecv@" + localhost + " notified semaphore - hostevent DOWN");
                        }
                    
                        if (name.equals("HBSELF") && backup) {
                            go = false;
                            break;
                        }
                        else continue;
                    }
                
                    /**
                    time = DateFormat.getTimeInstance(DateFormat.FULL);
                    msg = new String(buffer, 0, p.getLength());
                    trace.debug("hbrevc@" + localhost + " received: " 
                                                           + msg + " at " + time.format(new Date()));
                    **/
                
                    t1 = System.currentTimeMillis();
                    //time = DateFormat.getTimeInstance(DateFormat.FULL);
                    msg = new String(buffer, 0, p.getLength());
                    
                    if ((!firsthb) && name.equals("HBSELF")) {
                        //if (backup) {
                            //trace.debug("\nBACKUP " + name + ": hbrevc@" + localhost 
                                                  //+ " received heartbeat msg=(" + msg + ") from "
                                                  //+ recvhost + " at " + time.format(new Date()));   
                        //}  
                        //else {
                            //trace.debug("\nPRIMARY " + name + ": hbrevc@" + localhost 
                                                  //+ " received heartbeat msg=(" + msg + ") from "
                                                  //+ recvhost + " at " + time.format(new Date()));                            
                        //}
                    }
                    
                    if ((firsthb)  && (hbFromRecvHost(msg))) {
                    
                        trace.debug(name + ": hbrevc@" + localhost 
                                         + " received first heartbeat from "
                                         + recvhost + " at " + time.format(new Date()));
                            
                        s.setSoTimeout(hbint);
                        firsthb = false;
                        
                        event = recvhost + " UP";
                        trace.debug(name + ": hbrecv@" + localhost 
                                         + " will synchronize on semaphore - hostevent UP occurred");
                        synchronized (sem) {
                            trace.debug(name + ": hbrecv@" + localhost + " will notify semaphore"); 
                            sem.notifyAll();
                            trace.debug(name + ": hbrecv@" + localhost + " notified semaphore - hostevent UP");
                        }
                    }
                    i++;
                } while (true);
                
                if (name.equals("HBSELF") && backup) 
                    trace.debug("HBSELF: Recv thread closing socket and exiting");
                s.close();
            }
            catch (SocketException e) {
                trace.error(name + ": hbrecv could not create socket");
                System.exit(2);
            }
            catch (java.io.IOException e) {
                if (e.getMessage().equals("Receive timed out"))
                trace.warn(name + ": hbrecv missed heartbeat");
            }
        }   
    }
    
    private boolean hbFromRecvHost(String msg) {
        final int NUMTOKENS = 4;
        String token = null;
        int i;
        StringTokenizer st = new StringTokenizer(msg);
        if (st.countTokens() >= NUMTOKENS) {
            for (i=0; i < NUMTOKENS; i++) token = st.nextToken();
            if (recvhost.equals(token)) return (true);
        }
        return(false);
    }
    
    class HbSend extends java.lang.Thread {
        int realhbint = 0;
        DatagramSocket s;
        DatagramPacket p;
        int order = 0;
        
        /** Creates object HbSend */
        public HbSend() {                       
            this.realhbint = hbint / secfactor;
            trace.info("REAL HB INTERVAL = " + String.valueOf(realhbint));
            
            try {
                InetAddress address = InetAddress.getByName(sendhost);            
                s = new DatagramSocket();                            
                trace.debug(name + ": hbsend will connect ...");
                s.connect(address, portsend);
                trace.debug(name + ": hbsend connected!");
            }
            catch (UnknownHostException e) { 
                trace.error(name + ": unknow host ");
                System.exit(1);
            }
            catch (SocketException e) {
                trace.error(name + " - Exception in hbSend(): could not create socket. " + e.getMessage());
                System.out.println("ERROR: The Redundancy Manager could not open a necessary socket and is exiting." 
                    + "\n\tAnother EMS may already be started on this machine.");
                System.exit(2);
            }            
                
            trace.debug(name + ": hb send socket: bound to local port " 
                                                + Integer.toString(s.getLocalPort())
                                                + ", sending to port " + Integer.toString(s.getPort()));
        }

        private void sendHb() {
            String msg;
            byte[] data;
            
            try {
                order++;
                msg = "heartbeat " + String.valueOf(order) + " from " + localhost + "         ";
                data = msg.getBytes();
                p = new DatagramPacket(data, data.length);
                // if (order == 1) {
                if (!name.equals("HBSELF")) {
                    DateFormat time = DateFormat.getTimeInstance(DateFormat.FULL);
                    trace.debug(name + ": hbsend sending heartbeat \'" + msg 
                                     + "\" at " + time.format(new Date()));
                }
                s.send(p);
            }
            catch (IOException e) {
                trace.error(name + " - Send HB exception (could not send packet): " + e.getMessage());
                System.exit(3);
            }
            catch (Exception e) {
                trace.error(name + " - Send HB function got exception - " + e.getMessage());
                System.exit(4);
            }
        }
        
        public void run() {
            trace.debug(name + ": hbsend running - hbint = " + String.valueOf(hbint) +
                                           " - realhbint = " + String.valueOf(realhbint) +
                                           " - will spawn periodic background send thread");               
           
            for (;;) {
                if (go) sendHb(); 
                else {
                    trace.debug(name + " hbsend thread will exit ...");
                    break;
                }; 
                try {
                    sleep(realhbint);
                }
                catch (Exception e) {
                    trace.warn(name + ": hbsend thread sleep interrupted - " + e.getMessage());
                }
            }
        }   
    }
    
    
    public Heartbeat (boolean bootstart) throws UnknownHostException {
        
        // Save the name of the heartbeat object
        name = "HBSELF";
        
        if (!bootstart) backup = true;
        
        // Get localhost name
        localhost = InetAddress.getLocalHost().getHostName(); 
        recvhost = localhost;
        sendhost = localhost;
        
        trace.info("Heartbeat HBSELF being instantiated on host " + localhost);
        
        // Create semaphore object
        sem = new Object(); 
        trace.debug(name + ": semaphore instantiated");
                   
        // Get interval and security factor of the heartbeat
        if (!hbSelfGetPeriod()) {
            trace.warn(name + ": error in heartbeat configuration");
            initok = false;
            return;
        }
        
        if (!bootstart) {
            trace.debug(name + ": executed hbSelfGetPeriod()");
        }
            
        // Get port 
        String stport = RedMan.cfgtable.getProperty(name + "_PORT");
        if (stport != null) stport = stport.trim();
        else {
            trace.error(name + ": Invalid configuration file. Heartbeat port not specified");
            initok = false;
            return;
        }
        portrecv = Integer.parseInt(stport);
        if (backup) {
            portsend = portrecv;
            portrecv = portrecv - 1;
        }
        else {
            portsend = portrecv - 1;
        }
            
        if (backup) {
            trace.debug("HBSELF (backup) ports: recv = " + String.valueOf(portrecv)
                                                          + ", send = " + String.valueOf(portsend));
        }
        else {
            trace.debug("HBSELF (primary) ports: recv = " + String.valueOf(portrecv)
                                                           + ", send = " + String.valueOf(portsend));
        }
                        
        // Create the receiving thread
        hbrecv = new HbRecv();
        trace.info(name + ": heartbeat obj created thread obj hbrecv");
        hbrecv.setPriority(Thread.MAX_PRIORITY);
        
        // Create the sending thread
        hbsend = new HbSend(); 
        trace.info(name + ": heartbeat obj created thread obj hbsend");
        hbsend.setPriority(Thread.MAX_PRIORITY);
    }

    public void startHeartbeat() {
        if (!name.equals("HBSELF") && !torun) {
            trace.error("Heartbeat " + name + " is not supposed to run on this host (" 
                                     + localhost + ")"); 
            return;
        }
        if (!initok) {
            trace.error("Heartbeat " + name + " was not initialized correctly - Will not be started");
            return;
        }
        
        // Start the receiving thread
        trace.debug(name + ": starting thread hbrecv");
        hbrecv.start();       
        trace.info(name + ": heartbeat obj started thread hbrecv");
        
        // Start the sending thread
        trace.debug(name + ": starting thread hbsend");
        hbsend.start();        
        trace.info(name + ": heartbeat obj started thread hbsend"); 
        
        // Remember that this heartbeat object is running
        running = true;
    }
        
    
    private boolean hbSelfGetPeriod() {
        String stsecfactor, stperiod;
        
        // Get characterisitcs of the heartbeat
        stperiod = RedMan.cfgtable.getProperty(name + "_PERIOD");
        if (stperiod == null) {
            trace.error(name + ": Invalid configuration file. Heartbeat period not specified");
            return(false);
        }
        hbint = Integer.parseInt(stperiod);

        stsecfactor = RedMan.cfgtable.getProperty(name + "_SECURITY_FACTOR");
        if (stsecfactor == null) {
            trace.error(name + ": Invalid configuration file. Heartbeat security factor not specified");
            return(false);
        }
        secfactor = Integer.parseInt(stsecfactor);
        
        if (true) return(true);
        
        // Old code ...............Kept here just in case
        
        String hblist = RedMan.cfgtable.getProperty("HEARTBEATS");
        StringTokenizer st = new StringTokenizer(hblist);
        int intv=0, intv2=0, sf=0;
        
        hbint = 3000;
        secfactor = 3;
        trace.debug("HBSELF: hbint = " + Integer.toString(hbint) + ", secfact = 3");
        if (true) return(true);
        
        int i=1;
        while (st.hasMoreTokens()) {
            String hbname = st.nextToken();
            
            stperiod = RedMan.cfgtable.getProperty(hbname + "_PERIOD");   
            if (stperiod != null) stperiod = stperiod.trim();
            else {
                trace.error(name + ": Invalid configuration file. Heartbeat period not specified");
                return(false);
            }

            intv2 = Integer.parseInt(stperiod);
            if ((intv > intv2) || (i==1)) {
                intv = intv2;
                stsecfactor = RedMan.cfgtable.getProperty(hbname + "_SECURITY_FACTOR");
                if (stsecfactor != null) stsecfactor = stsecfactor.trim();
                else {
                    trace.error(name 
                        + ": Invalid configuration file. Heartbeat security factor not specified");
                    return(false);                    
                }
                sf = Integer.parseInt(stsecfactor);
            }
            i++;
        }
        
        // hbint = intv/sf;
        // secfactor = 2;
        // trace.debug("HBSELF: hbint = " + Integer.toString(hbint) + ", secfactor = 2"); 
        
        return(true);
    }
    
}